package com.gzhryc.socket.mosquitto;

import com.gzhryc.common.logging.Logger;
import com.gzhryc.socket.*;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;

import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * MQTT协议的客户端
 * @author 张浩
 * @version 0.1.3
 */
public class MosquittoThreadPoolMqttClient extends SocketNode{

    static Logger log = Logger.getLogger(MosquittoThreadPoolMqttClient.class);
    
    private ThreadPoolExecutor executor;
    private MainThread mainThread;
    private BlockingQueue<Runnable> blockQueue;
    
    private org.eclipse.paho.client.mqttv3.MqttClient mosquittoMqttClient;
    private org.eclipse.paho.client.mqttv3.MqttConnectOptions mosquittoConnectOptions;
    
    private SocketConfig mConfig;
    private SocketCallback mCallback;
    
    public MosquittoThreadPoolMqttClient(SocketConfig config,SocketCallback callback) {
    	this.mConfig = config;
    	this.mCallback = callback;
    	
    	this.blockQueue = new LinkedBlockingQueue<Runnable>();
    	this.executor = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaximumPoolSize()
    			,config.getKeepAliveTime(), TimeUnit.MILLISECONDS,this.blockQueue); 
    }
    
    public void start() {
        try {
            MqttClientPersistence e = null;
            if (StringUtils.isNotBlank(mConfig.getFileDir())) {
                e = new MqttDefaultFilePersistence(mConfig.getFileDir());
            } else {
                e = new MemoryPersistence();
            }
            this.mosquittoMqttClient = new org.eclipse.paho.client.mqttv3.MqttClient(mConfig.getUrl(), mConfig.getClientId(), e);
            if(mConfig.getWaitTime() != null) {
                this.mosquittoMqttClient.setTimeToWait(mConfig.getWaitTime());
            }
            this.mosquittoMqttClient.setCallback(new MosquittoClientMqttCallback(this,mosquittoMqttClient,mCallback));
            
            this.mosquittoConnectOptions = new org.eclipse.paho.client.mqttv3.MqttConnectOptions();
            if (StringUtils.isNotBlank(mConfig.getUserName())) {
                this.mosquittoConnectOptions.setUserName(mConfig.getUserName());
            }
            if (StringUtils.isNotBlank(mConfig.getUserPwd())) {
                this.mosquittoConnectOptions.setPassword(mConfig.getUserPwd().toCharArray());
            }
            this.mosquittoConnectOptions.setMqttVersion(org.eclipse.paho.client.mqttv3.MqttConnectOptions.MQTT_VERSION_3_1_1);
            this.mosquittoConnectOptions.setCleanSession(mConfig.isCleanSession());
            
            // 设置超时时间（秒）
            this.mosquittoConnectOptions.setConnectionTimeout(mConfig.getConnectionTimeout());  
            // 设置会话心跳时间（秒）
            this.mosquittoConnectOptions.setKeepAliveInterval(mConfig.getKeepAliveInterval());  
            
            //this.mosquittoConnectOptions.setAutomaticReconnect(true);  //自动链接
            
            final IMqttToken token = mosquittoMqttClient.connectWithResult(mosquittoConnectOptions);
            final MosquittoThreadPoolMqttClient self = this;
            if (token.isComplete()) {
            	runTask(new Runnable() {
            		@Override
            		public void run() {
            			SocketEvent event = new SocketEvent(self);
            			event.getSocketNode().setConnectedDate(new Date());
            			mCallback.active(event);
            		}
            	});
            }
        }catch(Exception e) {
			log.error(e.getMessage(),e);
			log.error("配置信息，{{0:json}}",mConfig);
        }
    }
    
    public void startService() {
    	if(mainThread == null || !mainThread.isAlive()) {
    		this.mainThread = new MainThread(this);
    		this.mainThread.setDaemon(true); //守护线程
    		this.mainThread.start();
    	}
    }
    
    public boolean isConnected(){
    	if(this.mosquittoMqttClient != null) {
    		return this.mosquittoMqttClient.isConnected();
    	}
    	return false;
    }
    
    public void runTask(Runnable runnable) {
    	this.executor.submit(runnable);
    }
    
    /**
     * 发布消息
     * @param message
     * @return
     */
    public boolean writeAndFlush(SocketMessage message) {
    	try {
    		if(isConnected()){
    			MqttMessage mqttMessage = new MqttMessage();
        		mqttMessage.setPayload(message.getBytes());
        		Object topic = message.getAttribute(SocketConstants.ATTR_MQTT_TOPIC);
        		Object qos = message.getAttribute(SocketConstants.ATTR_MQTT_QOS);
        		if(qos != null) {
        			mqttMessage.setQos(Integer.valueOf(qos.toString()));
        		}else {
        			mqttMessage.setQos(1);
        		}
        		
    			this.mosquittoMqttClient.publish(topic.toString(),mqttMessage);
    			return true;
    		}
			
		} catch (MqttPersistenceException e) {
			log.error(e.getMessage(), e);
		} catch (MqttException e) {
			log.error(e.getMessage(), e);
		}
    	return false;
    }
    
    /**
     * 订阅主题
     * @param topic
     * @param qos
     * @return
     */
    public boolean subscribe(String topic, int qos){
		try {
			if(isConnected()){
				this.mosquittoMqttClient.subscribe(topic,qos);
				return true;
			}
		} catch (MqttException e) {
			log.error(e.getMessage(),e);
		}
		return false;
    }
    
    /**
     * 取消订阅的主题
     * @param topic
     * @return
     */
    public boolean unsubscribe(String topic){
    	try {
    		if(isConnected()){
    			this.mosquittoMqttClient.unsubscribe(topic);
    			return true;
    		}
		} catch (MqttException e) {
			log.error(e.getMessage(),e);
		}
		return false;
    }
    
    /**
     * 重新连接
     */
    public void reconnect() {
    	if(mainThread != null) {
    		mainThread.interrupt();
    	}
    	
    	try {
	    	this.blockQueue.clear();  //清空任务队列
			if(this.mosquittoMqttClient.isConnected()){
				this.mosquittoMqttClient.disconnect();
	    	}
			this.mosquittoMqttClient.close();
	    }catch (MqttException e){
			log.error(e.getMessage(),e);
	    }
    	
    	if(mainThread != null) {
    		//以子线程方式重新启动
    		mainThread = new MainThread(this);
    		mainThread.setDaemon(true); //守护线程
    		mainThread.start();
    	}else {
    		start();
    	}
    }
    
    /**
     * 关闭连接并释放资源
     */
    public void close(){
    	try{
    		this.executor.shutdownNow(); //关闭线程池
    		this.blockQueue.clear();  //清空任务队列
			this.mosquittoMqttClient.disconnect();
    		this.mosquittoMqttClient.close();
    	}catch (MqttException e){
			log.error(e.getMessage(),e);
        }
    }
    
    public SocketConfig getConfig(){
    	return this.mConfig;
    }
    
    public static class MainThread extends Thread{
    	
    	final MosquittoThreadPoolMqttClient mqttClient;
    	
    	public MainThread(MosquittoThreadPoolMqttClient mqttClient) {
    		this.mqttClient = mqttClient;
    	}
    	
    	public void run() {
    		while(!mqttClient.isConnected()) {
    			try {
    				mqttClient.start();
    			}catch(Exception e) {
					log.error(e.getMessage(),e);
    			}
    			if(mqttClient.getConfig().getReconnectInterval() == null || mqttClient.getConfig().getReconnectInterval() == 0) {
					log.warn("未设置ReconnectInterval值");
    				return;
    			}
    			try {
    				Long millis = mqttClient.getConfig().getUnit().toMillis(mqttClient.getConfig().getReconnectInterval());
					Thread.sleep(millis);
				} catch (InterruptedException e) {
					log.error(e.getMessage(),e);
				}
    		}
    	}
    }

	@Override
	public String getId() {
		return mosquittoMqttClient.getClientId();
	}
}
